package com.sdk.es.http.kafka;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.sdk.es.http.conts.Conts;
import com.sdk.es.http.service.base.IEsBaseService;
import com.sdk.es.sdk.entity.common.EsBaseEntity;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;

import javax.annotation.Resource;
import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.List;

/**
 * @author wst
 * @date 2024/2/6
 *
 * 父类消息监听
 * @param <E>   具体索引实体
 * @desc 注：修改和删除业务极少，默认关闭监听，需要再打开
 */
@Slf4j
public abstract class EsBaseConsumer<E extends EsBaseEntity> {

    @Resource
    private IEsBaseService<E> baseService;

    protected Class<E> eClazz;

    private String saveTopic;
    private String updateTopic;
    private String deleteTopic;

    @PostConstruct
    private void init() {
        try {
            this.eClazz = (Class<E>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
            String indexName = initEsIndex();
            if (StrUtil.isNotBlank(indexName)) {
                saveTopic = String.format("%s%s%s", Conts.TOPIC_PREFIX, indexName, Conts.TOPIC_SAVE_SUFFIX);
                updateTopic = String.format("%s%s%s", Conts.TOPIC_PREFIX, indexName, Conts.TOPIC_UPDATE_SUFFIX);
                deleteTopic = String.format("%s%s%s", Conts.TOPIC_PREFIX, indexName, Conts.TOPIC_DELETE_SUFFIX);
            } else {
                log.error("当前方法未定义基础索引名");
            }
            log.debug("{} init:{}-{}", this.getClass().getSimpleName(), this.eClazz.getSimpleName(), indexName);
        } catch (Exception e) {
            log.error("获取参数实体类型异常：{}", e.getMessage(), e);
        }
    }

    /**
     * 初始化基础索引名称
     */
    protected abstract String initEsIndex();

    /**
     * 消息监听分发，子类需要注册 kafkaListener，监听具体 topic
     * @param messageList   具体消息体列表
     * @param topicList     队列 topic
     */
    protected void listener(@Payload List<String> messageList, @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topicList) {
        try {
            log.info("开始消费topic:{},消息数量:{}", topicList.get(0),messageList.size());
            List<E> entityList = new ArrayList<>();
            for(String message : messageList){
                E entity = JSONUtil.toBean(message, eClazz);
                entityList.add(entity);
            }
            if (topicList.get(0).equals(saveTopic)) {
                saveBatch(entityList);
            } else if (topicList.get(0).equals(updateTopic)) {
                update(entityList);
            } else if (topicList.get(0).equals(deleteTopic)) {
                delete(entityList);
            } else {
                log.warn("异步队列消息 topic 异常：{},消息列表：{}", topicList.get(0),messageList);
            }
        } catch (Exception e) {
            log.error("异步队列消息消费异常topic:{},消息列表：{}，异常：{}",topicList.get(0),messageList,e.getMessage(), e);
        }
    }

    protected void saveBatch(List<E> entityList) {
        try {
            baseService.saveBatch(entityList);
        } catch (Exception e) {
            log.error("kafka数据消费异常#topic={},message={}", saveTopic, entityList, e);
        }
    }

    protected void update(List<E> entityList) {
        try {
            for(E entity : entityList){
                baseService.update(entity);
            }
        } catch (Exception e) {
            log.error("kafka数据消费异常#topic={},message={}", updateTopic, entityList, e);
        }
    }

    protected void delete(List<E> entityList) {
        try {
            for(E entity : entityList){
                baseService.delete(entity);
            }
        } catch (Exception e) {
            log.error("kafka数据消费异常#topic={},message={}", deleteTopic, entityList, e);
        }
    }
}
